package org.momo.rpc.server.impl;


import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.momo.rpc.annotion.RateLimit;
import org.momo.rpc.protocol.RpcMessage;
import org.momo.rpc.protocol.RpcRequest;
import org.momo.rpc.protocol.RpcResponse;
import org.momo.rpc.protocol.constant.MessageStatus;
import org.momo.rpc.protocol.constant.MessageType;
import org.momo.rpc.provider.ServiceProvider;
import org.momo.rpc.provider.model.ProviderInfo;
import org.momo.rpc.ratelimit.RateLimiter;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;

@Slf4j
public class NettyServerHandler extends SimpleChannelInboundHandler<RpcMessage<Object>> {

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();
        log.debug("Client {}:{} open the channel", remoteAddress.getHostName(), remoteAddress.getPort());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();
        log.debug("Client {}:{} close the channel", remoteAddress.getHostName(), remoteAddress.getPort());
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcMessage<Object> reqMessage) throws Exception {
        RpcMessage.Header reqHeader = reqMessage.getHeader();
        RpcRequest request = (RpcRequest) reqMessage.getData();

        if(reqHeader.getStatus() == MessageType.HEARTBEAT_PING.getCode()) {
            RpcMessage<RpcResponse> pongMessage = RpcMessage.heartBeatResponse();
            ctx.writeAndFlush(pongMessage);
            return;
        }

        RpcResponse response;
        RpcMessage.Header resHeader = new RpcMessage.Header();
        resHeader.setId(reqHeader.getId());
        resHeader.setMessageType(MessageType.RESPONSE.getCode());

        // 反射获取调用的服务
        ProviderInfo providerInfo = ServiceProvider.getServiceClass(request.getServiceKey());
        Object serviceImpl = ServiceProvider.getServiceInstance(request.getServiceKey());

        if(providerInfo == null || serviceImpl == null) {
            response = RpcResponse.fail("service not found");
            resHeader.setStatus(MessageStatus.RESPONSE_ERROR.getCode());
        } else {
            Class<?> serviceClass = providerInfo.getServiceClass();
            try {
                // rate limit
                RateLimiter rateLimiter = providerInfo.getRateLimitMap().get(request.getMethodName());
                if(rateLimiter != null && !rateLimiter.limit()) {
                    throw new RuntimeException("rate limit");
                }

                Method method = serviceClass.getMethod(request.getMethodName(), request.getParamsClazz());
                Object result = method.invoke(serviceImpl, request.getParams());

                response = RpcResponse.success(result, method.getReturnType());
                resHeader.setStatus(MessageStatus.RESPONSE_SUCCESS.getCode());
            } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | RuntimeException e) {
                log.error("Server execute error! id: {}, service: {}, method: {}, error: {}", reqHeader.getId(), request.getServiceName(), request.getMethodName(), e.getMessage());
                response = RpcResponse.fail(e.getMessage());
                resHeader.setStatus(MessageStatus.RESPONSE_ERROR.getCode());
            }
        }

        // 封装response
        RpcMessage<RpcResponse> resMessage = new RpcMessage<>(resHeader, response);

        // 写入管道
        ctx.writeAndFlush(resMessage)
                .addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                        // ("Server write send message error! the message id is {}, the error is {}", resHeader.getId(), channelFuture.cause().getMessage());
                    }
                });
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        InetSocketAddress remoteAddress = (InetSocketAddress) ctx.channel().remoteAddress();

        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.READER_IDLE) {
                log.info("Client channel idle happen! Now close it, channel remote address: {}, port: {}", remoteAddress.getHostName(), remoteAddress.getPort());
                // ctx.close();
            }
        }
    }
}
